package com.syf.starter.soa.protocol.handle;

import com.syf.core.actuator.SyServiceDefinitionPublisher;
import com.syf.core.service.SyServiceDefinition;
import com.syf.core.service.context.ServiceRequest;
import com.syf.core.service.context.ServiceResponse;
import com.syf.core.service.invoke.ServiceInvokeCallback;
import com.syf.core.service.invoke.SyServiceInvoker;
import com.syf.kit.SyfJsonMapper;
import com.syf.kit.soa.core.SOAExchangeData;
import com.syf.kit.soa.core.SyfRpcMsg;
import com.syf.kit.soa.core.SyfSOAProtocolCont;
import com.syf.starter.SOAServerCont;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;


/**
 * 使用SyfServiceInvoker调用service
 */
@Slf4j
public class RpcCallServiceHandle extends SimpleChannelInboundHandler<SyfRpcMsg> {
    /**
     * 交易执行器
     */
    private final SyServiceInvoker serviceInvoker;

    /**
     * jsonMapper
     */
    private final SyfJsonMapper jsonMapper;

    /**
     * 交易执行的超时时间，单位：秒
     */
    private final int requestTimeoutSeconds;

    public RpcCallServiceHandle(SyServiceInvoker serviceInvoker, SyfJsonMapper jsonMapper, int requestTimeoutSeconds) {
        super();
        this.serviceInvoker = serviceInvoker;
        this.jsonMapper = jsonMapper;
        this.requestTimeoutSeconds = requestTimeoutSeconds;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, SyfRpcMsg syfRpcMsg) throws Exception {
        String serviceCode = syfRpcMsg.getServiceCode();

        if (!serviceInvoker.isExistService(serviceCode)) {
            ServiceResponse serviceResponse = buildErrorResponse(SOAServerCont.SERVICE_NOT_FOUND, "交易：" + serviceCode + "不存在");
            responseMsg(ctx, syfRpcMsg, serviceResponse);
            return;
        }

        ServiceRequest request;

        if (syfRpcMsg.getBizDataType() == SyfSOAProtocolCont.JSON_BIZ_DATA_TYPE) {
            SOAExchangeData soaExchangeData = jsonMapper.toBean(syfRpcMsg.getBizData(), SOAExchangeData.class);

            SyServiceDefinition serviceDefinition = SyServiceDefinitionPublisher.getInstance().getDefinitionByCode(syfRpcMsg.getServiceCode());
            request = new ServiceRequest(SOAServerCont.RPC_APP_COMPONENT_NAME);
            request.getHead().setTraceId(soaExchangeData.getTraceId());
            request.getHead().setCallCount(syfRpcMsg.getCallCount());
            request.getHead().setBizOrderNo(soaExchangeData.getBizOrderNo());
            request.getHead().setBizUseId(soaExchangeData.getBizUseId());
            request.getHead().setServiceDefinition(serviceDefinition);
            request.setDataObj(jsonMapper.toBean(soaExchangeData.getBizReqJson(), serviceDefinition.getRequestClass()));
        } else {
            ServiceResponse serviceResponse = buildErrorResponse(SOAServerCont.SYS_EXCEPTION_RESP_CODE, "业务数据，暂时只支持1-json");
            responseMsg(ctx, syfRpcMsg, serviceResponse);
            return;
        }

        if (syfRpcMsg.getReqType() == SyfSOAProtocolCont.REQUEST_RESPONSE_REQ_TYPE) { // request/response类请求
            final CountDownLatch timeoutLatch = new CountDownLatch(1);

            serviceInvoker.asyncInvokeService(serviceCode, request, new ServiceInvokeCallback<ServiceResponse>() {
                @Override
                public void onSuccess(ServiceResponse response) {
                    if (timeoutLatch.getCount() > 0) {
                        responseMsg(ctx, syfRpcMsg, response);
                        timeoutLatch.countDown();
                    }

                    // not to do，此处说明已经超时了
                }

                @Override
                public void onFailure(ServiceResponse response) {
                    if (timeoutLatch.getCount() > 0) {
                        responseMsg(ctx, syfRpcMsg, response);
                        timeoutLatch.countDown();
                    }
                }

                @Override
                public void throwable(Exception e) {
                    log.error("交易执行出现异常！", e);
                    if (timeoutLatch.getCount() > 0) {
                        ServiceResponse serviceResponse = buildErrorResponse(SOAServerCont.SYS_EXCEPTION_RESP_CODE, "交易执行出现异常：" + e.getMessage());
                        responseMsg(ctx, syfRpcMsg, serviceResponse);
                        timeoutLatch.countDown();
                    }
                }
            });

            // 交易执行超时处理
            final String traceId = request.getHead().getTraceId();
            ctx.channel().eventLoop().schedule(() -> {
                if (timeoutLatch.getCount() > 0L) {
                    log.error("交易链路:[{}],执行超时(配置的超时时间为 {} s)", traceId, requestTimeoutSeconds);
                    ServiceResponse serviceResponse = buildErrorResponse(SOAServerCont.SYS_EXCEPTION_RESP_CODE, "交易执行超时");
                    responseMsg(ctx, syfRpcMsg, serviceResponse);
                    timeoutLatch.countDown();
                }
            }, requestTimeoutSeconds, TimeUnit.SECONDS);
        } else if (syfRpcMsg.getReqType() == SyfSOAProtocolCont.TRIGGER_REQ_TYPE) { // trigger触发
            serviceInvoker.invokeService(serviceCode, request);
            responseMsg(ctx, syfRpcMsg, buildSuccessResponse());
        } else {
            // 不支持此请求类型
            ServiceResponse serviceResponse = buildErrorResponse(SOAServerCont.SYS_EXCEPTION_RESP_CODE, "暂不支持该请求类型");
            responseMsg(ctx, syfRpcMsg, serviceResponse);
        }
    }

    private ServiceResponse buildErrorResponse(String respCode, String respMsg) {
        return new ServiceResponse(respCode, respMsg);
    }

    private ServiceResponse buildSuccessResponse() {
        return new ServiceResponse();
    }

    private void responseMsg(ChannelHandlerContext ctx, SyfRpcMsg origRpcMsg, ServiceResponse serviceResponse) {
        String respJson = jsonMapper.toJsonString(serviceResponse);
        origRpcMsg.setBizData(respJson);
        ctx.writeAndFlush(origRpcMsg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (cause instanceof IOException) {
            log.warn("服务端[SYF-RPC-SERVER]发现：存在客户端连接：[{}]，异常中断！", ctx.channel().remoteAddress());
        } else {
            log.error("服务端[SYF-RPC-SERVER]发现异常", cause);
        }
    }
}
